package avicit.bdp.dds.server.worker.runner;


import avicit.bdp.common.utils.SpringApplicationContext;
import avicit.bdp.core.util.json.JSONUtils;
import avicit.bdp.dds.common.Constants;
import avicit.bdp.dds.common.thread.ThreadUtils;
import avicit.bdp.dds.common.utils.CommonUtils;
import avicit.bdp.dds.common.utils.FileUtils;
import avicit.bdp.dds.common.utils.LoggerUtils;
import avicit.bdp.dds.dispatch.enums.ExecutionStatus;
import avicit.bdp.dds.dispatch.enums.TaskType;
import avicit.bdp.dds.dispatch.model.TaskNode;
import avicit.bdp.dds.dispatch.process.Property;
import avicit.bdp.dds.dispatch.task.TaskTimeoutParameter;
import avicit.bdp.dds.remote.command.TaskExecuteResponseCommand;
import avicit.bdp.dds.server.entity.TaskExecutionContext;
import avicit.bdp.dds.server.worker.cache.TaskExecutionContextCacheManager;
import avicit.bdp.dds.server.worker.cache.impl.TaskExecutionContextCacheManagerImpl;
import avicit.bdp.dds.server.worker.processor.TaskCallbackService;
import avicit.bdp.dds.server.worker.task.AbstractTask;
import avicit.bdp.dds.server.worker.task.TaskManager;
import com.alibaba.fastjson2.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;


/**
 * task scheduler thread
 */
public class TaskExecuteThread implements Runnable {

    /**
     * logger
     */
    private final Logger logger = LoggerFactory.getLogger(TaskExecuteThread.class);

    /**
     * task instance
     */
    private TaskExecutionContext taskExecutionContext;

    /**
     * abstract task
     */
    private AbstractTask task;

    /**
     * task callback service
     */
    private TaskCallbackService taskCallbackService;

    /**
     * taskExecutionContextCacheManager
     */
    private TaskExecutionContextCacheManager taskExecutionContextCacheManager;

    private ExecutorService parseProcessOutputExecutorService;


    /**
     * constructor
     *
     * @param taskExecutionContext taskExecutionContext
     * @param taskCallbackService  taskCallbackService
     */
    public TaskExecuteThread(TaskExecutionContext taskExecutionContext, TaskCallbackService taskCallbackService) {
        this.taskExecutionContext = taskExecutionContext;
        this.taskCallbackService = taskCallbackService;
        this.taskExecutionContextCacheManager = SpringApplicationContext.getBean(TaskExecutionContextCacheManagerImpl.class);
        String threadLoggerInfoName = String.format(LoggerUtils.TASK_LOGGER_THREAD_NAME + "-%s", taskExecutionContext.getTaskAppId());
        parseProcessOutputExecutorService = ThreadUtils.newDaemonFixedThreadExecutor(threadLoggerInfoName, 3);
    }

    @Override
    public void run() {

        TaskExecuteResponseCommand responseCommand = new TaskExecuteResponseCommand(taskExecutionContext.getTaskInstanceId());
        // custom logger
        Logger taskLogger = LoggerFactory.getLogger(LoggerUtils.buildTaskId(LoggerUtils.TASK_LOGGER_INFO_PREFIX,
                taskExecutionContext.getProcessDefineId(),
                taskExecutionContext.getProcessInstanceId(),
                taskExecutionContext.getTaskInstanceId()));

        try {
            logger.info("script path : {}", taskExecutionContext.getExecutePath());
            TaskNode taskNode;
            //处理sparkFlow节点执行
            if(taskExecutionContext.getTaskType().equals(TaskType.SPARKFLOW.name())){
                taskNode = new TaskNode();
                taskNode.setName(taskExecutionContext.getTaskName());
                String execLocalPath = getExecLocalPath(taskExecutionContext);
                taskExecutionContext.setExecutePath(execLocalPath);

            }else{
                // task node
                 taskNode = JSONObject.parseObject(taskExecutionContext.getTaskJson(), TaskNode.class);
            }

            taskExecutionContext.setTaskParams(taskNode.getParams());
            taskExecutionContext.setEnvFile(CommonUtils.getSystemEnvPath());
            taskExecutionContext.setDefinedParams(getGlobalParamsMap());

            // set task timeout
            setTaskTimeout(taskExecutionContext, taskNode);

            taskExecutionContext.setTaskAppId(String.format("%s", taskExecutionContext.getTaskInstanceId()));

            task = TaskManager.newTask(taskExecutionContext, taskLogger);

            logger.info("task init...");
            printTaskLog(taskLogger, "开始运行---节点名称：" + taskNode.getName()+ "--节点ID：" + taskExecutionContext.getTaskInstanceId());
            // task init
            task.init();

            logger.info("task handle start......");
            printTaskLog(taskLogger, "节点任务处理开始");
            // task handle
            task.handle();
            logger.info("task handle end, start after method......");

            // task result process
            printTaskLog(taskLogger, "节点任务返回结果处理开始");
            task.after();

            logger.info("end after method, task instance id : {},task final status : {}",
                    taskExecutionContext.getTaskInstanceId(), task.getExitStatus());

            responseCommand.setStatus(task.getExitStatus().getCode());
            responseCommand.setEndTime(new Date());
            if (task.getProcessId() != null) {
                responseCommand.setProcessId(task.getProcessId());
            }
            responseCommand.setAppIds(task.getAppIds());

            if (taskExecutionContext.isNeedReturnResult()) {
                responseCommand.setResult(taskExecutionContext.getResult());
            }

            logger.info("task instance id : {},task final status : {}", taskExecutionContext.getTaskInstanceId(), task.getExitStatus());
            printTaskLog(taskLogger, "节点任务执行结束");
        } catch (Throwable e) {
            logger.error("task scheduler failure", e);
            printTaskLog(taskLogger, "节点任务执行失败:" + e);
            kill();
            responseCommand.setStatus(ExecutionStatus.FAILURE.getCode());
            responseCommand.setEndTime(new Date());
            responseCommand.setProcessId(task.getProcessId());
            responseCommand.setAppIds(task.getAppIds());
        } finally {
            try {
                taskExecutionContextCacheManager.removeByTaskInstanceId(taskExecutionContext.getTaskInstanceId());
                taskCallbackService.sendResult(taskExecutionContext.getTaskInstanceId(), responseCommand.convert2Command());
                //删除临时文件
                //FileUtils.deleteDir(taskExecutionContext.getExecutePath());
            } catch (Exception e) {
                logger.error("task scheduler finally failure", e);
                ThreadUtils.sleep(Constants.SLEEP_TIME_MILLIS);
                taskCallbackService.sendResult(taskExecutionContext.getTaskInstanceId(), responseCommand.convert2Command());
            }

            logger.info("task scheduler send result end, response command : {}", responseCommand);
        }

        parseProcessOutputExecutorService.shutdown();
    }

    /**
     * 增加统一的task日志输出
     *
     * @param taskLogger taskLogger
     */
    private void printTaskLog(final Logger taskLogger, String info) {
        parseProcessOutputExecutorService.submit(() -> taskLogger.info(info));
    }

    /**
     * get global paras map
     *
     * @return global paras map
     */
    private Map<String, String> getGlobalParamsMap() {
        Map<String, String> globalParamsMap = new HashMap<>(16);

        // global params string
        String globalParamsStr = taskExecutionContext.getGlobalParams();
        if (globalParamsStr != null) {
            List<Property> globalParamsList = JSONUtils.toList(globalParamsStr, Property.class);
            for (Property p : globalParamsList) {
                if (StringUtils.isNotBlank(p.getProp())) {
                    globalParamsMap.put(p.getProp(), p.getValue());
                }
            }
        }
        return globalParamsMap;
    }

    /**
     * set task timeout
     *
     * @param taskExecutionContext TaskExecutionContext
     * @param taskNode
     */
    private void setTaskTimeout(TaskExecutionContext taskExecutionContext, TaskNode taskNode) {
        // the default timeout is the maximum value of the integer
        taskExecutionContext.setTaskTimeout(Integer.MAX_VALUE);
        TaskTimeoutParameter taskTimeoutParameter = taskNode.getTaskTimeoutParameter();
        if (taskTimeoutParameter.getEnable()) {
            // get timeout strategy
            taskExecutionContext.setTaskTimeoutStrategy(taskTimeoutParameter.getStrategy().getCode());
            switch (taskTimeoutParameter.getStrategy()) {
                case WARN:
                    break;
                case FAILED:
                    if (Integer.MAX_VALUE > taskTimeoutParameter.getInterval() * 60) {
                        taskExecutionContext.setTaskTimeout(taskTimeoutParameter.getInterval() * 60);
                    }
                    break;
                case WARNFAILED:
                    if (Integer.MAX_VALUE > taskTimeoutParameter.getInterval() * 60) {
                        taskExecutionContext.setTaskTimeout(taskTimeoutParameter.getInterval() * 60);
                    }
                    break;
                default:
                    logger.error("not support task timeout strategy: {}", taskTimeoutParameter.getStrategy());
                    throw new IllegalArgumentException("not support task timeout strategy");

            }
        }
    }


    /**
     * kill task
     */
    public void kill() {
        if (task != null) {
            try {
                task.cancelApplication(true);
            } catch (Exception e) {
                logger.error(e.getMessage(), e);
            }
        }
    }


    /**
     * get execute local path
     *
     * @param taskExecutionContext taskExecutionContext
     * @return execute local path
     */
    private String getExecLocalPath(TaskExecutionContext taskExecutionContext) {
        return FileUtils.getProcessExecDir(taskExecutionContext.getProjectId(),
                taskExecutionContext.getProcessDefineId(),
                taskExecutionContext.getProcessInstanceId(),
                taskExecutionContext.getTaskInstanceId());
    }

}
